package com.seari.movedata.utils;

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.*;

import java.util.*;

public class influxdbJDBC {

    private static String openurl = PropertiesUtil.getProperty("commonConfig.properties", "openurl");
    private static String database = PropertiesUtil.getProperty("commonConfig.properties", "database");
    private static String measurement = PropertiesUtil.getProperty("commonConfig.properties", "measurement");


    private InfluxDB influxDB;

    public influxdbJDBC() {
        influxDbBuild();
        createRetentionPolicy();
    }

    /**连接时序数据库，获取influxDB**/
    public InfluxDB influxDbBuild() {
        influxDB = InfluxDBFactory.connect(openurl);
        Pong pong = influxDB.ping();
        if(pong != null){
            System.out.println("pong：" + pong + ",连接成功！");
        }else{
            System.out.println("连接失败");
            return null;
        }
        influxDB.createDatabase(database);
        return influxDB;
    }

    /*
     * 设置数据保存策略
     * defalut 策略名 /database 数据库名/ 30d 数据保存时限30天/ 1  副本个数为1/ 结尾DEFAULT 表示 设为默认的策略
     * */
    public void createRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT", "defalut",database,"30d",1);
        this.query(command);
    }

    /*
     * 查询
     * @param command 查询语句
     * @return
     * */
    public QueryResult query(String command) {
        Query query = new Query(command, database);
        QueryResult a = influxDB.query(query);
        return a;
    }

    /*
     * 插入
     * @param measurment 表
     * @param tags 标签
     * @param fields 字段
     * */
    public void insert(String measurement,Map<String, String> tags, Map<String, Object> fields) {
        Point.Builder builder = Point.measurement(measurement);//源码讲表名赋值到类中
        builder.tag(tags);//源码对标签进行验证，不得有空
        builder.fields(fields);//源码将数据值赋值到类中

        influxDB.write(database, "", builder.build());//向数据库写数据，builder.build()：验证表名不得为空，field数据长度大于0。
    }

    /*
     * 删除
     * @param command 删除语句
     * @return 返回错误信息
     */
    public String deleteMeasurementData(String command) {
        QueryResult result = influxDB.query(new Query(command, database));
        return result.getError();
    }

    /*
     * 创建数据库
     * @param dbName
     */
    public void creatDB(String dbName) {
        influxDB.createDatabase(dbName);
    }

    /*
     * 删除数据库
     * @param dbName
     */
    public void deleteDB(String dbName) {
        influxDB.deleteDatabase(dbName);
    }

    public static void main(String args[]){
        InfluxDBConnUtils influxDB = new InfluxDBConnUtils(openurl, database);
        InfluxDB client = InfluxDBFactory.connect(openurl,"admin","admin");


        List<Map<String,String>> abc = new ArrayList<Map<String,String>>();
        Map<String,String> map1 = new HashMap<String,String>();
        map1.put("tagid","1");
        map1.put("atime","a");
        map1.put("nvalue","b");
        Map<String,String> map2 = new HashMap<String,String>();
        map2.put("tagid","1");
        map2.put("atime","a");
        map2.put("nvalue","b");
        Map<String,String> map3 = new HashMap<String,String>();
        map3.put("tagid","1");
        map3.put("atime","a");
        map3.put("nvalue","b");
        Map<String,String> map4 = new HashMap<String,String>();
        map4.put("tagid","1");
        map4.put("atime","a");
        map4.put("nvalue","b");
        abc.add(map1);
        abc.add(map2);
        abc.add(map3);
        abc.add(map4);

        int i = 0;//控制循环，批量存储，每10万条数据存储一次
        //声明influxdb批量插入数据对象
        BatchPoints batchPoints = BatchPoints
                .database(database)
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();
        //遍历sqlserver获取数据
        for(Map<String, String> map : abc){
            //创建单条数据对象——表名
            Point point = Point.measurement(measurement)
                    //tag属性——只能存储String类型
                    .tag("TAG_TAGID", map.get("tagid"))
                    .tag("TAG_ATIME", map.get("atime"))
                    //field存储数据
                    .addField("FIELDS_NVALUE", map.get("nvalue"))
                    .build();
            //将单条数据存储到集合中
            batchPoints.point(point);
            i++;
            //每读取十万条数据提交到influxdb存储一次
            if(i / 2 == 1){
                i = 0;
                client.write(batchPoints);
                batchPoints = BatchPoints
                        .database(database)
                        .consistency(InfluxDB.ConsistencyLevel.ALL)
                        .build();
            }
        }
        client.write(batchPoints);
        System.out.println("结束时间：" + new Date().getTime() + ",i:");

    }
}
